package rocketmq;

import cn.hutool.json.JSONUtil;
import configer.SqlServerConfiger;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import utils.FormatData;

import java.io.UnsupportedEncodingException;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;

public class Producer {

    protected final static Logger logger = LoggerFactory.getLogger(Producer.class);

    private DefaultMQProducer producer;

    // nameServer节点IP地址和端口，多个用 ; 隔开
    public Producer(String producerGroup, String nameServer) {
        // 实例化消息生产者Producer
        producer = new DefaultMQProducer(producerGroup);

        // 设置NameServer的地址
        producer.setNamesrvAddr(nameServer);
        producer.setVipChannelEnabled(false);

        // 启动 producer 实例
        try {
            producer.start()/**
             * @Descriotion SqlServer数据库增量订阅
             */;
            logger.info("启动 producer 成功");
        } catch (MQClientException e) {
            logger.error("producer 启动失败");
        }
    }

    /**
     * @Descriotion 将SqlServer数据库全量数据发送到RocketMQ
     */
    public void sqlServerProduceSync(ResultSet resultSet, SqlServerConfiger sqlServerConfiger, String topic, String tag) {

        int columnCount;
        ResultSetMetaData md;
        FormatData formatData = new FormatData();
        String[] splitIndexField = sqlServerConfiger.getDataIndexs().split("\\,");
        try {
            md = resultSet.getMetaData();
            columnCount = md.getColumnCount();
            logger.info("输出 Result <" + resultSet + ">");
            do {
                formatData.clean();
                for (int i = 1; i <= columnCount; i++) {
                    // 过滤RowNumber字段
                    if ("RowNumber".equals(md.getColumnName(i))) {
                        continue;
                    }
                    formatData.putBody(md.getColumnName(i), resultSet.getString(i));
                }

                formatData.setDataUser(sqlServerConfiger.getDataUser());
                formatData.setDataType(sqlServerConfiger.getDataType());
                formatData.setDataName(sqlServerConfiger.getDataName());
                formatData.setDataLevel(sqlServerConfiger.getDataLevel());
                formatData.setDataCrypt(sqlServerConfiger.getDataCrypt());

                for (int i = 0; i < splitIndexField.length; i++) {
                    formatData.putDataIndex(splitIndexField[i], resultSet.getString(splitIndexField[i]));
                }

                Message msg = null;
                try {
                    msg = new Message(topic, tag, formatData.getKeys(),
                            JSONUtil.toJsonStr(formatData.toJSON()).getBytes(RemotingHelper.DEFAULT_CHARSET));
                } catch (UnsupportedEncodingException e) {
                    logger.error("RocketMQ.Message 编码失败");
                }
                logger.info("全量数据格式化 " + formatData.toJSON());
                logger.info("消息组装 " + msg.toString());
                try {
                    SendResult sendResult = producer.send(msg);
                    logger.info("消息发送结果 " + sendResult);
                } catch (MQClientException e) {
                    logger.error("MQClient 出错 { " + e + " }");
                } catch (RemotingException e) {
                    logger.error("Remoting 出错 { " + e + " }");
                } catch (MQBrokerException e) {
                    logger.error("MQBroker 出错 { " + e + " }");
                } catch (InterruptedException e) {
                    logger.error("Interrupted 出错 { " + e + " }");
                }
            } while (resultSet.next());
        } catch (SQLException throwables) {
            logger.info("输出 Result <" + resultSet + "> 失败");
        }
    }


    /**
     * @Descriotion 将SqlServer数据库增量数据发送到RocketMQ
     */
    public String sqlServerProduceSub(ResultSet resultSet, SqlServerConfiger sqlServerConfiger, String topic, String tag) {

        int columnCount;
        ResultSetMetaData md;
        FormatData formatData = new FormatData();
        String lsn = null;
        String[] splitIndexField = sqlServerConfiger.getDataIndexs().split("\\,");
        try {
            md = resultSet.getMetaData();
            columnCount = md.getColumnCount();
            logger.info("输出 Result <" + resultSet + ">");
            do {
                formatData.clean();
                // 从结果集中选取需要的数据(__$operation为“2”代表插入，__$operation为“4”代表执行更新操作后的值)
                if ("4".equals(resultSet.getString("__$operation")) ||
                        "2".equals(resultSet.getString("__$operation"))) {
                    lsn = resultSet.getString("__$start_lsn");
                    for (int i = 1; i <= columnCount; i++) {
                        // 过滤部分字段
                        if ("__$seqval".equals(md.getColumnName(i)) || "__$update_mask".equals(md.getColumnName(i))
                                || "__$operation".equals(md.getColumnName(i)) || "__$start_lsn".equals(md.getColumnName(i))
                                || "__$end_lsn".equals(md.getColumnName(i))) {
                            continue;
                        }
                        formatData.putBody(md.getColumnName(i), resultSet.getString(i));
                    }
                    formatData.setDataUser(sqlServerConfiger.getDataUser());
                    formatData.setDataType(sqlServerConfiger.getDataType());
                    formatData.setDataName(sqlServerConfiger.getDataName());
                    formatData.setDataLevel(sqlServerConfiger.getDataLevel());
                    formatData.setDataCrypt(sqlServerConfiger.getDataCrypt());

                    for (int i = 0; i < splitIndexField.length; i++) {
                        formatData.putDataIndex(splitIndexField[i], resultSet.getString(splitIndexField[i]));
                    }

                    Message msg = null;
                    try {
                        msg = new Message(topic, tag, formatData.getKeys(),
                                JSONUtil.toJsonStr(formatData.toJSON()).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    } catch (UnsupportedEncodingException e) {
                        logger.error("RocketMQ.Message 编码失败");
                    }
                    logger.info("增量数据格式化 " + formatData.toJSON());
                    logger.info("消息组装 " + msg.toString());
                    try {
                        SendResult sendResult = producer.send(msg);
                        logger.info("消息发送结果 " + sendResult);
                    } catch (MQClientException e) {
                        logger.error("MQClient 出错 { " + e + " }");
                    } catch (RemotingException e) {
                        logger.error("Remoting 出错 { " + e + " }");
                    } catch (MQBrokerException e) {
                        logger.error("MQBroker 出错 { " + e + " }");
                    } catch (InterruptedException e) {
                        logger.error("Interrupted 出错 { " + e + " }");
                    }
                }
            } while (resultSet.next());
        } catch (SQLException throwables) {
            logger.info("输出 Result <" + resultSet + "> 失败");
        }
        return lsn;
    }


    // 如果不再发送消息，关闭Producer实现
    public void close() {
        producer.shutdown();
    }

}